package com.mall.seckill.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateOrderRequest;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.seckill.cache.CacheManager;
import com.mall.seckill.dal.entitys.PromoItem;
import com.mall.seckill.dal.persistence.PromoItemMapper;
import com.mall.seckill.dto.SeckillRequest;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.zookeeper.Transaction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @author lichao
 * @createdate 2021-02-01 21:46
 */
@Component
@Slf4j
public class MqMessageProducer {

    @Autowired
    PromoItemMapper promoItemMapper;
    @Autowired
    CacheManager cacheManager;

    private TransactionMQProducer producer;


    @PostConstruct
    public void init(){
        //创建实例
        producer = new TransactionMQProducer("seckill_producer");

        //设置监听地址
        producer.setNamesrvAddr("127.0.0.1:9876");

        //事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {

                String transactionId = message.getTransactionId();
                cacheManager.setCache(transactionId,"init",1);

                Map map = (Map) o;
                PromoItem promoItem = (PromoItem) map.get("promoItem");
                //执行本地事务
                try {
                    Integer effectedRows = promoItemMapper.decreaseStock(promoItem.getPsId(),promoItem.getItemId());
                    if (effectedRows < 1) {
                        log.error("库存扣减失败");
                        cacheManager.setCache(transactionId,"fail",1);
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                    cacheManager.setCache(transactionId,"success",1);
                    return LocalTransactionState.COMMIT_MESSAGE;

                }catch (Exception e){
                    log.error("库存扣减时发生异常");
                    cacheManager.setCache(transactionId,"unknown",1);
                    return LocalTransactionState.UNKNOW;
                }

            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {

                //检查本地事务结果
                String transactionId = messageExt.getTransactionId();
                String value = cacheManager.checkCache(transactionId);
                if(value.equals("success")){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                if(value.equals("fail")){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public Boolean sendPromoTranactionMessage(HashMap<String, Object> request){
        Message message = new Message();
        message.setTopic("promo_order");

        message.setBody(JSON.toJSONString(request).getBytes(Charset.forName("utf-8")));

        TransactionSendResult sendResult = null;

        try {
            //1、发送half消息 2、返回half 3、执行本地事务
            sendResult =producer.sendMessageInTransaction(message,request);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //half消息返回失败
        if(sendResult == null){
            return false;
        }

        LocalTransactionState localTransactionState = sendResult.getLocalTransactionState();

        if(localTransactionState.equals(LocalTransactionState.COMMIT_MESSAGE)){
            return true;
        }
        return false;
    }

}
